package com.bw.gmall.realtime.app.dws;

import com.bw.gmall.realtime.bean.ShopNewVisitOrderBean;
import com.bw.gmall.realtime.bean.ShopNotVisitOrderBean;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/*
*
* dws  1.  关连  把多个事实表连载一起       开窗        关联维度
*
* */
public class ShopNotPurchasedVisitOrderApp {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql("" +
                "CREATE TABLE Zb2_Shop_Not_purchased_User_order (\n" +
                "  us STRING COMMENT '用户ID',\n" +
                "  shops STRING COMMENT '店铺',\n" +
                "  dt STRING COMMENT '日期',\n" +
                "  snv STRING COMMENT '新访客标识（根据您的查询推测）',\n" +
                "  rt TIMESTAMP(3) COMMENT '事件时间',\n" +
                "  total_amount DOUBLE COMMENT '订单金额',\n" +
                "    WATERMARK FOR rt AS rt - INTERVAL '5' SECOND\n" +  // 添加水位线定义
                ") " + MyKafkaUtil.getKafkaDDL("Zb2_Shop_Not_purchased_User_order", "Zb2_Shop_Not_purchased_User_order"));
//        tableEnv.executeSql("select * from Zb2_Shop_Not_purchased_User_order").print();

        Table uv = tableEnv.sqlQuery("SELECT \n" +
                "    shops AS shop,\n" +
                "    COUNT(DISTINCT us) AS hfnum,\n" +
                "    COUNT(DISTINCT CASE WHEN total_amount > 0 THEN us END) AS hfcjnum,\n" +
                "    COUNT(DISTINCT us)  - COUNT(DISTINCT CASE WHEN total_amount > 0 THEN us END) AS hfwcjnum,\n" +
                "    ROUND(COUNT(DISTINCT CASE WHEN total_amount > 0 THEN us END) * 100.0 / COUNT(DISTINCT us), 2) AS zfzhl,\n" +
                "    ROUND(SUM(CASE WHEN total_amount > 0 THEN total_amount ELSE 0 END) / NULLIF(COUNT(DISTINCT CASE WHEN total_amount > 0 THEN us END), 0), 2) AS kdj,\n" +
                "    cast(TUMBLE_START(rt, INTERVAL '1' HOUR)as String) AS start_time,\n" +
                "    cast(TUMBLE_END(rt, INTERVAL '1' HOUR)  as String) AS end_time\n" +
                "FROM Zb2_Shop_Not_purchased_User_order where snv = '0'  \n" +
                "GROUP BY shops, TUMBLE(rt, INTERVAL '1' HOUR)");
//        uv.execute().print();

        DataStream<Tuple2<Boolean, ShopNotVisitOrderBean>> tuple2DataStream = tableEnv.toRetractStream(uv, ShopNotVisitOrderBean.class);

        SingleOutputStreamOperator<ShopNotVisitOrderBean> map = tuple2DataStream.map(t -> t.f1);
        map.print(">>>>>>>");
//////         将Table转换为DataStream
//////         写入ClickHouse
        map.addSink(MyClickHouseUtil.getSinkFunction("insert into table shop_notvisit_order_stats_zb1 values(?,?,?,?,?,?,?,?)"));

        //
        env.execute("ShopNewVisitOrderApp");
    }
}